#include "app_pool.h"
#include "log.h"
#include "mqueue.h"
#include "pthread.h"
#include <stdlib.h>
#include "errno.h"

static mqd_t mqd;
static pthread_t threads[THREAD_NUM];

// 每个线程要循环作的事情：1.循环从队列中获取任务，2.执行任务
static void *app_pool_worker_func(void *arg)
{
    log_info("app_pool thread start...");
    while (1)
    {
        Task task;

        mq_receive(mqd, (char *)&task, sizeof(Task), 0);

        task.task_func(task.arg);
    }
}

int app_pool_init(void)
{
    // 0.清理旧的队列文件
    mq_unlink(MQ_NAME);

    // 1.创建队列
    struct mq_attr attr;
    attr.mq_maxmsg = MQ_MAXSIZE;
    attr.mq_msgsize = sizeof(Task);
    mqd = mq_open(MQ_NAME, O_CREAT | O_RDWR, 0666, &attr);

    if (mqd == -1)
    {
        perror("mqd_open failed...");
        mq_unlink(MQ_NAME);
        return -1;
    }

    // 2.初始化线程数组
    for (size_t i = 0; i < THREAD_NUM; i++)
    {
        int ret = pthread_create(&threads[i], NULL, app_pool_worker_func, NULL);
        if (ret != 0)
        {
            perror("pthread_create failed...");
            mq_unlink(MQ_NAME);
            return -1;
        }
    }

    return 0;
}

int app_pool_add_task(void *(*task_func)(void *arg), void *arg)
{
    Task *task = (Task *)malloc(sizeof(Task));
    task->task_func = task_func;
    task->arg = arg;

    int ret = mq_send(mqd, (char *)task, sizeof(Task), 0);
    if (ret == -1)
    {
        perror("mq_send failed...");
        free(task);
        return -1;
    }
    return 0;
}

int app_pool_destory(void)
{
    for (int i = 0; i < THREAD_NUM; i++)
    {
        pthread_cancel(threads[i]);
        pthread_join(threads[i],NULL);
    }

    // 关闭消息队列
    mq_close(mqd);
    // 删除消息队列
    mq_unlink(MQ_NAME);
    return 0;
}
